package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

public class DosAbnorBolt extends AbstractBolt{


	private static final long serialVersionUID = 1L;
		
	private Map<String,Integer> freqMap = new HashMap<String,Integer>();
//	private HashMap<String,Integer> dipPInfo = new HashMap<String,Integer>();
	private Map<String,String> tupleMap = new HashMap<String,String>();
//	private Map<String,Integer> streampkgMap = new HashMap<String,Integer>();
//	private Map<String,Integer> streamoctetsMap = new HashMap<String,Integer>();
	private int accessNums = 60;	
//	private DosSliding cache;

	public DosAbnorBolt(int accessFres){
		this.accessNums = accessFres;
//		cache = new DosSliding(3);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d","dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour", "abntype", "idcno", "sourcecountry","streamoctets","streampacket"));
		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport","accesstime","sourcearea",
				"sourcegeo","sourcecountry","destarea","destgeo","evaluate","abnormal","desc","attnum","delaytime","flag"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60);
		return conf;
	}

	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{
			String url = tuple.getStringByField("url").trim();	
			String type =  tuple.getStringByField("type").trim();	
			if(url.isEmpty()){ //TCP
				
				String dip = tuple.getStringByField("dip");
				String dport = tuple.getStringByField("dport");
				String sip = tuple.getStringByField("sip");			
				String sport = tuple.getStringByField("sport");
				String gis = new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
				String idc = tuple.getStringByField("idcno");				
				String tripleKey = dip + "|" + dport + "|" + sip + "|" + type + "|" + idc;
				
				if(freqMap.containsKey(tripleKey)) {
					freqMap.put(tripleKey, freqMap.get(tripleKey) + 1);
				}else{
					freqMap.put(tripleKey, 1);
				}
				tupleMap.put(tripleKey, dip + "|" + dport + "|" + sip + "|" + sport +"|" + gis +  "|"  + type + "|"  + idc);
			}
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}

	//定时发送
	private void emitCountingData(OutputCollector collector) {
		
		try{
			HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
//			SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmm");		
//			Calendar c = Calendar.getInstance();
//			String dataMin = sdf.format(c.getTime());
//			c.add(Calendar.MINUTE, -1);// 1 minute ago
//			String dateMin1 = sdf.format(c.getTime());
//			c.add(Calendar.MINUTE, -1);// 2 minutes ago
//			String dateMin2 = sdf.format(c.getTime());
	
//			dipPInfo = new HashMap<String,Integer>();
			
			
//			String info = String.format("######DEBUG:DDOS MAP[%d] T[%s] T1[%s] T2[%s]", 
//					freqMap.size(),	dataMin , dateMin1 , dateMin2);
//			Logger.getRootLogger().info(info);
			
			for(String dkey : freqMap.keySet()) {	
			
				if(freqMap.get(dkey) >= accessNums){				
//					dipPInfo.put(dkey + "|" + dataMin, freqMap.get(dkey));
//					Map<Integer,HashMap<String,Integer>> infoMap = cache.getTuplesWindow();			
					
//					info = String.format("######DEBUG:DDOS Slid Window Map[%d]", 
//							infoMap.size());
//					Logger.getRootLogger().info(info);
					
					
//					if(infoMap != null ) {
//						int num = 0;
//						for(HashMap<String,Integer> singleMap: infoMap.values()) {
//							
//							if(singleMap != null) {
//								if(singleMap.containsKey(dkey+"|" + dateMin1) || singleMap.containsKey(dkey+"|" + dateMin2)){								
//									num++;								
//								}
//							}
//						}
						
//						Logger.getRootLogger().info(String.format("######DEBUG:DDOS NUM>60 n[%d] %s" , num, dkey));
						
//						if(num >= 2){
							if(tupleMap.containsKey(dkey)) {
								String[] tuplesp = tupleMap.get(dkey).split("\\|",-1);
								System.out.println("target >>>>>>>>> "+tupleMap.get(dkey) + " >>>>>>>>> frequency:" +  freqMap.get(dkey));
								
								String dip = tuplesp[0];
								String dport = tuplesp[1];
								String sip = tuplesp[2];
								String sport = tuplesp[3];
								String gis = tuplesp[4];
								String type = tuplesp[5];
								String idc = tuplesp[6];
								String[] gisArray = gis.split("#");
								String destAreaName = gisArray[0];
								String destGis = gisArray[1];
								String sourceAreaName = gisArray[2];
								String sourceGis = gisArray[3];
								String sourceAreaCountry = gisArray[4];
								String sourceAreaId = gisArray[5];
								String sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
								String sourceAreaCityId = gisArray[7];
								String sourceAreaProvinceId = gisArray[8];
								
								
								String desc = "DDoS Attack";
								String colname = "DOS";
								String ABRNORMAL = "2";
//								if(type.equals("1"))
//								{
//									desc = "TCP flood";
//									colname = "DOS";
//									//ABRNORMAL = "6";
//								}
//								else if(type.equals("2"))
//								{
//									desc = "UDP flood";
//									colname = "DOS";
//									//ABRNORMAL = "7";
//								}
//								else
//								{
//									desc = "TCP flood ";
//									colname = "DOS";
//								}
								
								String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis()+15000));
								String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_2";	
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEIP", sip);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTPORT", dport);							
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ABRNORMAL", ABRNORMAL);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTIP", dip);							
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTGEO", destGis);							
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", freqMap.get(dkey)+"");
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "EVALUATE", "40");	
								if(sourceProvinceName.equals("")) sourceProvinceName = sourceAreaCountry;
								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);
//								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS",String.valueOf(streamoctetsMap.get(dkey)));	
//								hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(streampkgMap.get(dkey)));	
								
								
								String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
						    	String rowkey3 = String.format("%s_%s", accesstime, sip);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "ACCESSTIME", accesstime);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEIP", sip);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",sourceGis));
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "ATTNUM", String.valueOf(freqMap.get(dkey)));
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "AREAID", sourceAreaProvinceId);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTIP", dip);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTPORT", dport);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "TYPE", "2");
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
						    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);
						    	
								Date current = new Date();
								SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
								SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
								SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
						    	String dateStr_h = sdf_h.format(current);
								String dateStr_d = sdf_d.format(current);
								String dateStr_hour = sdf_hour.format(current);
								
								collector.emit(new Values(dip,sip,dport,sport,dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
						    			sourceProvinceName, freqMap.get(dkey), dateStr_hour,colname , idc, sourceAreaCountry,0L,0L));
						    	
								
							}
//						}
//					}
				}
			}
//			if(dipPInfo.size() > 0){
//				cache.addPInfo(dipPInfo);
//			}
//			cache.advanceHead();
		}catch(Exception e){
			
		}finally{
//			dipPInfo.clear();
			freqMap.clear();
			tupleMap.clear();
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}


	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			
			if(isTickTuple(tuple)) {
				emitCountingData(collector);  
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("DDosAbnorBolt throws an exception!");
		}	
	}
	

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
	}

}
